package com.atguigu.reactor;

import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.SignalType;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author Jungle
 * @create 2024-01-09 22:27
 */
@Slf4j
public class FluxDemo {

    public static void main(String[] args) throws IOException {
        FluxDemo demo = new FluxDemo();
        // demo.thread();
        demo.thread1();
        // demo.dispose();
        // demo.create();
        // new FluxDemo().handler();
        System.in.read();
        // demo.generate();
        // fluxDemo.buffer();
        // fluxDemo.limit();

        // request(N)：找发布者请求N次数据  总共能得到N*bufferSize 个数据

    }

    public void thread1() {
        Scheduler scheduler = Schedulers.newParallel("parallel-scheduler", 4);

        // 流的每一个操作产生新流，产生新发布者
        Flux<String> flux = Flux.range(1, 2)
                .map(i -> 10 + 1)
                .log()
                .publishOn(scheduler)
                .log()
                .map(i -> "value " + i) ;

        // 只要不指定线程池，默认发布者的线程就是订阅者的线程
        new Thread(()->flux.subscribe(System.out::println)).start();
    }

    public void thread() {
        // 响应式编程：全异步 、消息 、事件回调
        // 流的发布、中间操作，默认使用当前线程
        Flux.range(1, 10)
                .publishOn( Schedulers.parallel())   //在哪个线程池把这个流的数据和操作执行了
                .map(i->i+10)
                .log()
                .map(integer -> integer + 10)
                // .subscribeOn(Schedulers.single())
                // .log()
                .subscribe();


        // publish:改变发布者所在线程池
        // subscribe0n:改变订i阅者所在的线程池


        // 调度器 ：线程池
        Schedulers.immediate();// 默认  无执行上下文  当前线程运行所有操作
        Schedulers.single();  //使用固定的一个单线程
        Schedulers.boundedElastic();  //有界，弹性调度 ; 不是无限扩充的线程池；线程池中有10*CPU核心个线程 ，队列默认100K  keepAliveTime 60s
        Schedulers.parallel();
        Schedulers.fromExecutor(new ThreadPoolExecutor(4, 8, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(1000)));
    }


    public void handler() {
        Flux.range(1, 10)
                .handle((value, sink) -> {
                    // SynchronousSink 只能调用一个next
                    System.out.println("拿到的值" + value);
                    sink.next("张三：" + value); //可以向下发送数据的通道
                    // sink.next("李四：" + value); //这里只能调用一个
                })
                .log()
                .subscribe()
        ;
    }




    // 异步环境下
    public void create() {
        Flux.create(fluxSink -> {
            MyListener myListener = new MyListener(fluxSink);
            for (int i = 0; i < 100; i++) {
                myListener.onLine("张:" + i);
            }
        }).log().subscribe();


    }

    class MyListener {
        FluxSink<Object> sink;

        public MyListener(FluxSink<Object> sink) {
            this.sink = sink;
        }

        // 用户登录  触发onLine监听
        public void onLine(String userName) {
            System.out.println("用户登录了：" + userName);
            sink.next(userName); //传入用户
        }
    }

    public void dispose() {
        Flux<Integer> flux = Flux.range(1, 10000)
                .delayElements(Duration.ofSeconds(1))
                .map(i -> i + 7)
                .log();

        // 1、消费者是实现了Disposable可取消
        Disposable disposable = flux.subscribe(System.out::println);

        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(10);
                disposable.dispose();  //销毁
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "t1").start();
    }


    // 以编程方式创建序列
    // Sink： 接收器，水槽，通道
    // source :数据源  sink :接受端
    public void generate() {
        Flux<Object> flux = Flux.generate(() -> 0,  //初始state值
                (state, sink) -> {
                    // 0-10
                    if (state <= 10) {
                        sink.next(state);  //把元素传出去
                    } else {
                        sink.complete();  //完成信号
                    }


                    if (state == 7) {
                        sink.error(new RuntimeException("我不喜欢7"));
                    }
                    return state + 1;  //返回新的迭代state值
                });
        flux.log()
                .doOnError(throwable -> System.out.println("throwable = " + throwable))
                .subscribe();

    }

    public void limit() {
        Flux.range(1, 1000)
                .log()  //限流触发看上游是怎么限流获取数据的
                .limitRate(100)  //一次预取30个元素
                .subscribe();

        // 75% 预取策略   limitRate(100)
        // 第一次抓取100个数据，如果75%的元素已经处理了，继续抓取新的75%元素;  如果是7 *0.75=5.25  一次取6个
    }

    public void buffer() {
        Flux<List<Integer>> buffer = Flux.range(1, 10)
                .buffer(3).log();//缓冲区只能缓存3个元素 消费者最多一次可以拿到3个元素

        // 10个元素，buffer（3） 消费者请求4次，数据消费完成
        //缓冲区（批处理）
        // 消费者每次request(1) 拿到的是几个真正的数据：buffer的数据
        // buffer.subscribe(v -> {
        //     System.out.println("类型 = " + v.getClass() + "\t值 = " + v);
        //     for (Integer integer : v) {
        //         System.out.println("查询订单：" + integer);
        //     }
        // });

        // 这个只会一个一个的发
        // Flux.range(1, 10).subscribe(System.out::println);


        buffer.subscribe();

    }


    /**
     * subscribe:订阅流：没订阅之前流什么也不做
     * 流的元素开始流动，发生数据变化
     * 响应式编程： 数据流(Flux+Mono)+变化传播（操作）
     */

    public static void customSubscribe(String[] args) {
        // onErrorXXX doOnXXXX
        // doOnxxx: 发生这个事件的时候产生一个回调，通知你（不能改变）
        // onXxx: 发生这个事情后执行一个动作，可以改变元素，信号
        //A0P:普通通知（前置、后置、异常、返回)   环绕通知(ProceedingJoinPoint )
        Flux<String> flux = Flux.range(1, 10)
                .map(i -> {
                    // System.out.println("map..."+i);
                    if (i == 9) {
                        // i = 10 / (9 - i); //数学运算异常  doOnXXX
                    }
                    return "哈哈" + i;

                })
                // .onErrorComplete()  //流错误的时候，把错误吃掉，转为正常信号
                ;
        // flux.subscribe();  //流被订阅
        // flux.subscribe(v -> System.out.println("v = " + v)); //指定订阅规则：正常消费者，只是消费正常元素
        // flux.subscribe(
        //         v -> System.out.println("v = " + v),  //正常情况 （正常情况和异常情况只能走一个）
        //         throwable -> System.out.println("throwable = " + throwable), //异常情况
        //         () -> System.out.println("流结束了。。。。") //感知正常结束
        // );


        // 流的生命周期钩子，可以传播给订阅者
        flux.subscribe(new BaseSubscriber<String>() {
            // 生命周期钩子，订阅关系绑定的时候触发
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                //   流被订阅的时候触发
                System.out.println("绑定了》》》" + subscription);
                //     找发布者要数据
                //     request(1);  //要一个数据
                requestUnbounded();  //要无限数据
            }

            // 每个元素出发onNext
            @Override
            protected void hookOnNext(String value) {
                System.out.println("数据到达:正在处理：" + value);
                // if (value.equals("哈哈5")) {
                //     cancel();  //取消流
                // }
                // request(1);
            }

            // hookOnComplete/ hookOnError 二选一执行
            @Override
            protected void hookOnComplete() {
                System.out.println("流正常结束。。。。");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("流异常了。。。。" + throwable);
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("流被取消了。。。");
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("最终回调。。。。一定会被执行");
            }
        });

    }


    public static void FluxMono(String[] args) {
        // Flux.concat(Flux.just(1, 2, 3), Flux.just(7, 8, 9)).subscribe(System.out::println);
        Flux.range(1, 7)
                // .log() //日志 onNext(1-7)
                .filter(i -> i > 3)
                // .log() //日志
                .map(i -> "haha-" + i)
                .log()  //onNext(haha-4-7)
                .subscribe(System.out::println);
    }

    /**
     * 响应式编程的核心：看懂弹珠图
     * 信号：、正常/异常(取消、错误)
     * SignalType：
     * SUBSCRIBE:被订阅
     * REQUEST: 请求了N个元素
     * CANCEL： 流被取消
     * ON_SUBSCRIBE: 在订阅的时候
     * ON_NEXT： 在元素到达时候
     * ON_ERROR： 在流错误
     * ON_COMPLETE: 在流正常完成时
     * AFTER_TERMINATE： 中断以后
     * CURRENT_CONTEXT: 当前是上下文
     * ON_CONTEXT 感知上下文
     * <p>
     * <p>
     * doOnxxx Api触发时机
     * 1、doNext：每个数据（流的数据）到达时候出发
     * 2、doOnEach: 每个元素（流的数据和信号）到达的时候触发
     * 3、doOnRequest : 消费者请求流元素的时候
     * 4、doOnError：在流发生错误的时候
     * 5、doOnSubScribe: 在流被订阅的时候
     * 6、doOnTerminate: 发送取消/异常信号中断了流
     * 7、doOnCancle： 流被取消
     * 8、doOnDiscard: 流中元素被忽略的时候
     */
    public static void doOnxx(String[] args) {
        // 关键 doOnNext:表示流中某个元素到达以后触发一个回调
        // doOnxxx要感知某个流的事件，写在这个流的后面，新流的前面
        Flux.just(1, 2, 3, 4, 5, 6, 7, 0, 5, 6)
                .doOnNext(e -> System.out.println("元素到达==>" + e))
                .doOnEach(integerSignal -> {
                    System.out.println("doOnEach" + integerSignal);
                })
                .map(e -> 10 / e)
                .doOnError(throwable -> {
                    System.out.println("数据库已经保存了此异常：+" + throwable);
                })
                .doOnNext(e -> System.out.println("元素到哈==>" + e))
                .subscribe(System.out::println);
    }

    /**
     * Mono<Integer>  :只有一个Integer
     * Flux<Integer>  : 有很多Integer
     */
    public static void fluxDoOn(String[] args) throws IOException {

        // Mono<Integer> just = Mono.just(1);
        // just.subscribe(System.out::println);

        // 空流：链式API中，下面的操作符，操作的是上面的流
        // 事件感知:当流发生什么事的时候，触发—个回调;系统调用提前定义好的钩子函数(Hook【钩子函数】) ;  do0nXxx ;
        Flux<Integer> flux = Flux.range(1, 7)

                .delayElements(Duration.ofSeconds(1))
                .doOnComplete(() -> {
                    log.info("==>{}", "流正常结束。。。。");
                })
                .doOnCancel(() -> {
                    log.info("==>{}", "流已经被取消");
                })
                .doOnError(throwable -> log.info("流出错了==>{}", throwable))
                .doOnNext(c -> {
                    log.info("doOnNext==>{}", c);
                });
        //有一个信号，此时代表完成信号
        flux.subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                log.info("订阅者和发布者绑定好了==>{}", subscription);
                request(1); //要一个元素
            }

            @Override
            protected void hookOnNext(Integer value) {
                log.info("元素到达==>{}", value);
                if (value < 5) {
                    request(1);
                    if (value == 3) {
                        int i = 10 / 0;
                    }
                }
            }

            @Override
            protected void hookOnComplete() {
                log.info("==>{}", "数据流结束");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                log.info("==>{}", "数据流异常了");
            }

            @Override
            protected void hookOnCancel() {
                log.info("==>{}", "数据流被取消");
            }

            @Override
            protected void hookFinally(SignalType type) {
                log.info("结束信号==>{}", type);
            }
        });


        // flux.subscribe(System.out::println);

        //@formatter:off  线程暂停一会儿
        try { TimeUnit.SECONDS.sleep( 2 ); } catch (InterruptedException e) { e.printStackTrace();}
        //@formatter:on


        // Flux<Integer> range = Flux.range(1, 7);
        // // 下面两个新流不一样
        // Flux<Integer> filter = range.filter(e -> e > 2);
        // Flux<Integer> filter1 = range.filter(e -> e > 5);

        System.in.read();

    }

    /**
     * 测试Flux
     */
    public static void flux() throws IOException {
        // Mono  0|1 个元素的流
        // Flux:N 个元素的流
        // 发布者发布数据流：源头

        Flux<Integer> just = Flux.just(1, 2, 3, 4, 5);

        // 流不消费就没用：消费：订阅
        just.subscribe(e -> System.out.println("e1=" + e));
        // 一个流可以对应多个消费者
        just.subscribe(e -> System.out.println("e2=" + e));
        //对于每个消费者来说流都是一样的;广播模式;

        System.out.println("====================");

        Flux<Long> flux = Flux.interval(Duration.ofSeconds(1));//每秒产生一个从0开始的递增数字
        flux.subscribe(System.out::println);
        System.in.read();


    }
}
